package com.example.mq.sender;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.example.mq.enums.MsgStatusEnum;
import com.example.mq.po.MsgDelayPO;
import com.example.mq.po.MsgPO;
import com.example.mq.service.MsgDelayService;
import com.example.mq.service.MsgService;
import com.example.utils.CollUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class MqSendRetryJob implements DisposableBean
{
    @Resource
    private MsgService msgService;
    @Resource
    private IMsgSender msgSender;
    @Resource
    private MsgDelayService messageDelayService;

    private volatile boolean stop;

    /*
     * 每20秒执行一次
     */
    @Scheduled(fixedDelay = 20 * 1000)
    public void sendRetry()
    {
        while (true)
        {
            log.info("开始执行发送重试任务...");
            /**
             * 查询出需要重试的消息，重新投递
             * select * from t_msg where status = 0 or (status = 2 and send_retry = 1 and next_retry_time <= 当前时间)
             */
            Page<MsgPO> page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            LambdaQueryWrapper<MsgPO> qw = Wrappers.lambdaQuery(MsgPO.class)
                    .eq(MsgPO::getStatus, MsgStatusEnum.INIT.getStatus())
                    .or(lq -> lq.eq(MsgPO::getStatus, MsgStatusEnum.FAIL.getStatus())
                            .eq(MsgPO::getSendRetry, 1)
                            .le(MsgPO::getNextRetryTime, LocalDateTime.now()));
            this.msgService.page(page, qw);
            //如果查询出来的为空 || 当前服务要停止了(stop=true)，则退出循环
            if (CollUtils.isEmpty(page.getRecords()) || stop)
            {
                break;
            }
            //投递重试
            for (MsgPO msgPO : page.getRecords())
            {
                this.msgSender.sendRetry(msgPO);
            }
        }
    }



    @Override
    public void destroy() throws Exception
    {
        this.stop = true;
    }




    /*
     * 每分钟执行一次
     */
    @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.MINUTES)
    public void sendDelayRetry()
    {

        /**
         * 查询出需要重试的消息（状态为0 and 期望投递时间 <= 当前时间 + 2分钟） || （投递失败的 and 需要重试 and 下次重试时间小于等于当前时间 + 2分钟）
         * select * from t_msg where ((status = 0 and expect_send_time<=当前时间+2分钟) or (status = 2 and send_retry = 1 and next_retry_time <= 当前时间 + 2分钟))
         */
        LocalDateTime time = LocalDateTime.now().plusMinutes(2);
        LambdaQueryWrapper<MsgDelayPO> qw = Wrappers.lambdaQuery(MsgDelayPO.class)
                .and(query -> query.and(lq ->
                                lq.eq(MsgDelayPO::getStatus, MsgStatusEnum.INIT.getStatus())
                                        .le(MsgDelayPO::getExpectSendTime, time))
                        .or(lq -> lq.eq(MsgDelayPO::getStatus, MsgStatusEnum.FAIL.getStatus())
                                .eq(MsgDelayPO::getSendRetry, 1)
                                .le(MsgDelayPO::getNextRetryTime, time)));
        qw.orderByAsc(MsgDelayPO::getId);

        // 先获取最小的一条记录的id
        MsgDelayPO msgDelayPO =  messageDelayService.getOne(qw);
        if (msgDelayPO == null)
            return;

        this.msgSender.sendDelayRetry(msgDelayPO);

        String minMsgId = msgDelayPO.getId();

        // 循环中继续向后找出id>minMsgId的所有记录，然后投递重试
        while (true)
        {
            //select * from t_msg where ((status = 0 and expect_send_time<=当前时间+2分钟) or (status = 2 and send_retry = 1 and next_retry_time <= 当前时间 + 2分钟)) and id>#{minMsgId}
            qw = Wrappers.lambdaQuery(MsgDelayPO.class)
                    .and(query -> query.and(lq ->
                                    lq.eq(MsgDelayPO::getStatus, MsgStatusEnum.INIT.getStatus()).le(MsgDelayPO::getExpectSendTime, time))
                            .or(lq -> lq.eq(MsgDelayPO::getStatus, MsgStatusEnum.FAIL.getStatus()).eq(MsgDelayPO::getSendRetry, 1)
                                    .le(MsgDelayPO::getNextRetryTime, time)));
            qw.gt(MsgDelayPO::getId, minMsgId);
            qw.orderByAsc(MsgDelayPO::getId);

            Page<MsgDelayPO> page = new Page<>();
            page.setCurrent(1);
            page.setSize(500);
            this.messageDelayService.page(page, qw);

            // 如果查询出来的为空 || 当前服务要停止了(stop=true)，则退出循环
            if (CollUtils.isEmpty(page.getRecords()) || stop)
            {
                break;
            }

            // 投递重试
            for (MsgDelayPO msgPO : page.getRecords())
            {
                this.msgSender.sendDelayRetry(msgPO);
            }

            // minMsgId = 当前列表最后一条消息的id
            minMsgId = page.getRecords().get(page.getRecords().size() - 1).getId();
        }
    }





}
